
#include "ae.h"
#include "buffer.h"
#include "session.h"
#include "thread.h"
#include "mylog.h"
#include "zmalloc.h"

#include <stdio.h>
#include <string.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <errno.h>

#define READ_BUFFER_INITIAL									1024
#define SEND_BUFFER_INITIAL									1024

static long long nextSessionId = 0;

typedef enum writeState{
	ERROR_WRITE	=	-1,
	OPEN_WRITE		= 0,
	CLOSE_WRITE	= 1
} writeState;

void closeWriteStateSession(struct session* session) {
	if (!session) {
		mylog(LOG_WARNING, "%s, %d", __FUNCTION__, __LINE__);
		return;
	}

	struct thread* t = session->t;
	if (!t) {
		mylog(LOG_WARNING, "%s, %d", __FUNCTION__, __LINE__);
		return;
	}

	aeDeleteFileEvent(t->threadEventLoop, session->cfd, AE_WRITABLE);
	session->sendEventState = CLOSE_SENDEVENT;
	retrieveAllBuffer(session->sendBuffer);
}

void onWriteSession(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask) {
	struct session* session = (struct session*) clientData;
	if (!session) {
		return;
	}

	if (session->cfd < 0) {
		mylog(LOG_WARNING, "in onWrieteLeftBuf session cfd < 0!!!");
		return;
	}

	mylog(LOG_INFO, "onWriteLeftBuf, session id[%d]!!!", session->id);
	int sendComplete = 0;
	while(1) {
		int sendLen = 0;
		sendLen = send(session->cfd, getReadIndexBuffer(session->sendBuffer), getReadableBytesBuffer(session->sendBuffer), 0);
		if (sendLen > 0) {
			moveReadIndexBuffer(session->sendBuffer, sendLen);
			if (getReadableBytesBuffer(session->sendBuffer) <= 0) {
				sendComplete = 1;
				break;
			}
		} else if (sendLen == -1) {
			if (errno == EAGAIN) {
				break;
			} else {
				mylog(LOG_DEBUG, "in on write left buf error!!!, session id[%d], errno[%d]", session->id, errno);
				break;
			}
		} else {
			mylog(LOG_DEBUG, "in on write left buf error, session id[%d]!!!", session->id);
			break;
		}
	}

	if (sendComplete == 1) {
		closeWriteStateSession(session);
		if (session->callBackHandle) {
			if (session->callBackHandle->onWriteCompleteConnectionCallBack) {
				session->callBackHandle->onWriteCompleteConnectionCallBack(session);
			}
		}
	}
}

writeState isOpenWriteSession(struct session* session) {
	if (!session) {
		return ERROR_WRITE;
	}

	if (session->sendBuffer && session->sendEventState == OPEN_SENDEVENT) {
		return OPEN_WRITE;
	}

	return CLOSE_WRITE;
}

void openWriteStateSession(struct session* session) {
	if (!session) {
		mylog(LOG_WARNING, "%s, %d", __FUNCTION__, __LINE__);
		return;
	}

	struct thread* t = session->t;
	if (!t) {
		mylog(LOG_WARNING, "%s, %d", __FUNCTION__, __LINE__);
		return;
	}

	aeCreateFileEvent(t->threadEventLoop, session->cfd, AE_WRITABLE, onWriteSession, session);
	session->sendEventState = OPEN_SENDEVENT;
}

struct session* initSession() {
	struct session* session = (struct session* )zcalloc(sizeof(struct session));
	if (!session) {
		mylog(LOG_WARNING, "zcalloc session error!!!");
		return NULL;
	}

	session->readBuffer = initBuffer(READ_BUFFER_INITIAL);
	if (!session->readBuffer) {
		mylog(LOG_WARNING, "zmalloc session read buffer error!!!");
		zfree(session);
		return NULL;
	}

	session->id = nextSessionId++;
	session->cfd = INVALID_FD;
	session->inUse = NO_USE;
	session->sendEventState = CLOSE_SENDEVENT;
	mylog(LOG_INFO, "init session id[%d]", session->id);
    return session;
}

void destroySession(struct session* session) {
	if (!session) {
		return;
	}

	mylog(LOG_INFO, "destroy session id[%d]", session->id);
	struct thread* t = session->t;
	if (!t) {
		return;
	}

	if (session->cfd) {
		if (session->sendEventState == OPEN_SENDEVENT) {
			aeDeleteFileEvent(t->threadEventLoop, session->cfd, AE_WRITABLE);
			session->sendEventState = CLOSE_SENDEVENT;
		}
		aeDeleteFileEvent(t->threadEventLoop, session->cfd, AE_READABLE);
		close(session->cfd);
	}

	if (session->readBuffer) {
		destroyBuffer(session->readBuffer);
	}

	if (session->sendBuffer) {
		destroyBuffer(session->sendBuffer);
	}

	if (session->conAddr) {
		zfree(session->conAddr);
	}

	zfree(session);
}
// no thread safe!!!
int readSession(struct session* session) {
	if (!session) {
		return -1;
	}

	int badRead = 1;
	int readLen = 0;
	while (1) {
		int readBufferSize = getWriteableBytesBuffer(session->readBuffer);
		if (readBufferSize <= 0) {
			int ret = expandBuffer(session->readBuffer);
			if (ret != 0) {
				mylog(LOG_WARNING, "%s, %d, session id[%d], readBuffer expand is error!!!", __FUNCTION__, __LINE__, session->id);
				break;
			}
			readBufferSize = getWriteableBytesBuffer(session->readBuffer);
		}
		int rlen = recv(session->cfd, getWriteIndexBuffer(session->readBuffer), readBufferSize, 0);
		mylog(LOG_DEBUG, "%s, %d, in read session, session id[%d], recv len[%d]", __FUNCTION__, __LINE__, session->id, rlen);
		if (rlen > 0) {
			moveWriteIndexBuffer(session->readBuffer, rlen);
			readLen += rlen;
		}else if (rlen == 0) {
			mylog(LOG_INFO, "%s, %d, session id[%d] be close!!!", __FUNCTION__, __LINE__, session->id);
			break;
		} else if (rlen == -1) {
			if (errno == EAGAIN) {
				mylog(LOG_DEBUG, "%s, %d, session id[%d], have some data not read!!!", __FUNCTION__, __LINE__, session->id);
				break;
			} else {
				mylog(LOG_DEBUG, "%s, %d, session id[%d] error!!!", __FUNCTION__, __LINE__, session->id);
				badRead = 0;
				break;
			}
		} else {
			mylog(LOG_DEBUG, "%s, %d, session id[%d]", __FUNCTION__, __LINE__, session->id);
			badRead = 0;
			break;
		}
	}

	mylog(LOG_DEBUG, "%s, %d, session id[%d], read len[%d]", __FUNCTION__, __LINE__, session->id, readLen);
	if (readLen == 0 || badRead == 0) {
		return 0;
	}

	return readLen;
}

int writeSessionEx(struct session* session, char* sendBuf, int sendBufLen) {
	if (!session) {
		return -1;
	}

	sendMsgInThread(session, sendBuf, sendBufLen);
}

int writeSession(struct session* session, char* sendBuf, int sendBufLen) {
	if (!session || !sendBuf || sendBufLen < 0) {
		mylog(LOG_WARNING, "%s, %d", __FUNCTION__, __LINE__);
		return -1;
	}

	char* buf = sendBuf;
	int hasSendBytes = 0;
	if (isOpenWriteSession(session) == CLOSE_WRITE) {
		int bufStartPos = 0;
		int bufEndPos = sendBufLen;
		while (1) {
			int writeLen;
			writeLen = send(session->cfd, buf + bufStartPos, bufEndPos -  bufStartPos, 0);
			if (writeLen > 0) {
				bufStartPos += writeLen;
				hasSendBytes += writeLen;
				if (bufEndPos <= bufStartPos) {
					mylog(LOG_DEBUG,  "session id[%d], session ip[%s], session port[%d], buf start pos[%d], buf end pos[%d]", session->id, session->cip, session->cport, bufStartPos, bufEndPos);
					break;
				}
			} else if (writeLen == -1) {
				if (errno == EAGAIN) {
					break;
				} else {
					mylog(LOG_DEBUG, "send buf error!!!, session id[%d], session ip[%s], session port[%d] errno[%d]!!!", session->id, session->cip, session->cport, errno);
					return -3;
				}
			} else {
				mylog(LOG_DEBUG,  "send buf error!!!, session id[%d]!!!", session->id);
				return -4;
			}
		}
	}

	if (hasSendBytes < sendBufLen) {
		appendBuffer(session->sendBuffer, (char*)sendBuf + hasSendBytes, sendBufLen - hasSendBytes);
		if (isOpenWriteSession(session) == CLOSE_WRITE ) {
			openWriteStateSession(session);
		}
	} else {
		if (session->callBackHandle) {
			if (session->callBackHandle->onWriteCompleteConnectionCallBack) {
				session->callBackHandle->onWriteCompleteConnectionCallBack(session);
			}
		}
	}
	return 0;
}

void shutdownSession(struct session* session) {
	if (!session) {
		return;
	}

	mylog(LOG_INFO, "shutdown session id[%d]", session->id);
	struct thread* t = session->t;
	if (!t) {
		return;
	}

	if (session->cfd) {
		if (session->sendEventState == OPEN_SENDEVENT) {
			aeDeleteFileEvent(t->threadEventLoop, session->cfd, AE_WRITABLE);
			session->sendEventState = CLOSE_SENDEVENT;
		}
		aeDeleteFileEvent(t->threadEventLoop, session->cfd, AE_READABLE);
		//shutdown(session->cfd, SHUT_WR);
		close(session->cfd);
	}

	if (session->readBuffer) {
		destroyBuffer(session->readBuffer);
		session->readBuffer = NULL;
	}

	if (session->sendBuffer) {
		destroyBuffer(session->sendBuffer);
		session->sendBuffer = NULL;
	}

	zfree(session);
}
